package org.apache.rocketmq.proxy.test.rpc;

import apache.rocketmq.v2.MessagingServiceGrpc;
import apache.rocketmq.v2.QueryRouteRequest;
import apache.rocketmq.v2.QueryRouteResponse;
import com.google.common.util.concurrent.ListenableFuture;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder;
import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import io.grpc.stub.MetadataUtils;

import javax.net.ssl.SSLException;
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

public class RpcClient {

    private static final Duration KEEP_ALIVE_DURATION = Duration.ofSeconds(30);
    private static final int GRPC_MAX_MESSAGE_SIZE = Integer.MAX_VALUE;

    private final Endpoints endpoints;
    private final ManagedChannel channel;
    private final MessagingServiceGrpc.MessagingServiceFutureStub futureStub;
    private final MessagingServiceGrpc.MessagingServiceBlockingStub blockingStub;
    private final MessagingServiceGrpc.MessagingServiceStub stub;

    private long activityNanoTime;

    @SuppressWarnings("deprecation")
    public RpcClient(Endpoints endpoints) throws SSLException {
        this.endpoints = endpoints;
        final SslContextBuilder builder = GrpcSslContexts.forClient();
        builder.trustManager(InsecureTrustManagerFactory.INSTANCE);
        SslContext sslContext = builder.build();

        final NettyChannelBuilder channelBuilder =
                NettyChannelBuilder.forTarget(endpoints.getGrpcTarget())
                        // .withOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                        .keepAliveTime(KEEP_ALIVE_DURATION.toNanos(), TimeUnit.NANOSECONDS)
                        .maxInboundMessageSize(GRPC_MAX_MESSAGE_SIZE)
                        .intercept(LoggingInterceptor.getInstance())
                        .sslContext(sslContext);
        // Disable grpc's auto-retry here.

        final List<InetSocketAddress> socketAddresses = endpoints.toSocketAddresses();
        if (null != socketAddresses) {
            final IpNameResolverFactory ipNameResolverFactory = new IpNameResolverFactory(socketAddresses);
            channelBuilder.nameResolverFactory(ipNameResolverFactory);
        }

        this.channel = channelBuilder.build();
        this.futureStub = MessagingServiceGrpc.newFutureStub(channel);
        this.blockingStub = MessagingServiceGrpc.newBlockingStub(channel);
        this.stub = MessagingServiceGrpc.newStub(channel);
        this.activityNanoTime = System.nanoTime();
    }


    public Duration idleDuration() {
        return Duration.ofNanos(System.nanoTime() - activityNanoTime);
    }


    public void shutdown() throws InterruptedException {
        channel.shutdown().awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
    }


    //根据主题信息向服务端查询路由信息的方法
    public void queryRoute(Metadata metadata, QueryRouteRequest request, Executor executor, Duration duration) {
        this.activityNanoTime = System.nanoTime();
        final ListenableFuture<QueryRouteResponse> future = futureStub
                .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata))
                .withExecutor(executor)
                .withDeadlineAfter(duration.toNanos(), TimeUnit.NANOSECONDS)
                .queryRoute(request);
        future.addListener(() -> {
            try {
                QueryRouteResponse queryRouteResponse = future.get();
                System.out.println(queryRouteResponse);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, executor);
    }

}
